package cn.azzhu.day06;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;

import java.io.RandomAccessFile;

/**
 * Tuple2<String,String>
 *     文件,数据
 * @author azzhu
 * @create 2020-09-20 00:06:04
 */
public class MyParFileSource extends RichParallelSourceFunction<Tuple2<String,String>> {
    private String path;

    private boolean flag = true;

    /**
     * @param path  /var/data
     */
    public MyParFileSource(String path) {
        this.path = path;
    }

    public MyParFileSource(){}
    //只执行一次
    @Override
    public void run(SourceContext<Tuple2<String, String>> ctx) throws Exception {
        final int subtaskIndex = getRuntimeContext().getIndexOfThisSubtask();
        final RandomAccessFile randomAccessFile = new RandomAccessFile(path + "/" + subtaskIndex + ".txt", "r");

        while (flag) {
            String line = randomAccessFile.readLine();
            if(line != null) {
                line = new String(line.getBytes("ISO-8859-1"),"UTF-8");
                //数据发送出去
                ctx.collect(Tuple2.of(subtaskIndex + "",line));
            } else {
                //没有数据
                Thread.sleep(1000);
            }
        }
    }

    @Override
    public void cancel() {
        flag = false;
    }
}
